| Python cdp ( Chrome DevTools Protocol ) 爬虫 | 您所在的位置：网站首页 › network python 库 › Python cdp ( Chrome DevTools Protocol ) 爬虫 |
Python cdp ( Chrome DevTools Protocol ) 爬虫
|
devtools-protocol：https://github.com/ChromeDevTools/devtools-protocol Chrome DevTools Protocol：https://chromedevtools.github.io/devtools-protocol/ 实现 CDP 协议的库：https://github.com/ChromeDevTools/awesome-chrome-devtools cdp chrome：https://github.com/search?q=cdp+chrome&type=repositories&p=2 1、关于 Chrome Debugging ProtocolChrome 提供了 websocket 调试接口用于对当前 Tab 内页面的 DOM、网络、性能、存储 等等进行调试，我们常用的开发者工具就是基于此接口，这个接口也支持远程调用，在启动参数中加上 --remote-debugging-port=9222 即可。
远程调试接口是一个 WebSocket 的接口，Chrome 提供的开发者工具是一种客户端，自己写代码调用也是一种客户端。DevTools 可以 attach 到一个远程运行的 Chrome 实例，来进行 debug， 启动一个 host Chrome 实例：chrome.exe --remote-debugging-port=9222指定一个用户的 profile 启动一个 client 客户端 Chrome 实例：chrome.exe --user-data-dir=在客户端中输入 http://localhost:9222，你会看到当前 client 端的 DevTools 就像内嵌在电脑中Chrome 的 devtools 一样，然后你就可以通过 client 操作他了。 当用 client 连接到 9222 的时候，DevTools 前端 会被 host Chrome 实例 serve 为一个远程服务端的 web application，他会通过 HTTP 协议 fetch HTML，JavaScript，CSS，一旦加载，DevTools 会和 host 建立一个 ws 链接，并开始交换 JSON 信息也可以用自己的实现来替代 DevTools 前端，除了 localhost:9222， 远程调试还提供了一个JSON接口，用于管理浏览器的 Tab 页面。（ localhost:9222/json 来获取到 ws 通信的 JSONobject，并修改使用它们， HTTP Endpoints 章节中有详细内容。 ） 2、实战 Chrome Headless 数据抓取实战 Chrome Headless 数据抓取（上）：https://blog.csdn.net/chixulu6723/article/details/100730003 启动 Chrome启动之前，需要关闭所有 Chrome 窗口 Linux 中的 screen 命令使用：https://blog.csdn.net/han0373/article/details/81352663 如果在远程服务器上建议在 screen 里运行，一个小工具防止网络突然中断：$ screen -S chrome 然后会打开一个新的 shell，可以用 Ctrl + A + D 切出来，或者断开SSH 直接切出来。再进去只需要执行：$ screen -r chrome 然后在 screen 里面的 shell 执行（ 本机 Windows 调试把 google-chrome-unstable 换成chrome.exe ）：$ google-chrome-unstable --headless --remote-debugging-port=9222 --user-data-dir='/home/luke/chrome-data/baidu' --user-agent='Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3080.5 Safari/537.36' 解释下参数： --headless：无头模式，就是无界面模式运行--remote-debugging-port：开启远程调试，端口9222和我们之前转发出来的端口一致--user-data-dir：设置独立的文件保存目录，建议一个网站一个目录--user-agent：伪装浏览器，默认User-Agent里的浏览器叫HeadlessChrome，很容易被发现 连接调试端口启动之前，需要关闭所有 Chrome 窗口，然后在启动， 远程调试接口是一个 WebSocket 的接口，Chrome 提供的开发者工具是一种客户端，我们自己写代码调用也是一种客户端。这里先用开发者工具测试，后面会写代码来实现。 在本地浏览器开：http://服务器IP:9222/json ，本机测试的话就是 http://127.0.0.1:9222/json 。我已经用 SSH 把服务器的 9222 转发到本机的 9222 了，这是效果：
这里列出了当前远程浏览器内打开的Tab，每个页面一个UUID用以识别。已知接口： http://127.0.0.1:9222/json ：查看已经打开的Tab列表http://127.0.0.1:9222/json/version : 查看浏览器版本信息http://127.0.0.1:9222/json/new?http://www.baidu.com : 新开Tab打开指定地址http://127.0.0.1:9222/json/close/ac5a6adb-bb53-44f1-a9e6-2354bd724924 : 关闭指定Tabhttp://127.0.0.1:9222/json/activate/69301801-d503-42a3-9335-3e448a780857 : 切换到目标Tab 抓取百度远程打开，新开一个 Tab 打开百度首页，然后刷新 http://127.0.0.1:9222/json ，可以看到百度已经打开了：
注意到，有一个devtoolsFrontendUrl，那就是开发者工具的前端地址，就是一个html应用，url里面传过去WebSocket调试地址。打开这个地址就可以看到熟悉的开发者工具了！注意：这个窗口调试的是远程chrome上的页面。
如果你想看看页面在远程服务器的 Chrome 里渲染的结果，在开发者工具里切换到 Performance，勾选 Screenshots，点刷新图标，重新加载完成就可以看到逐帧加载的截图。 远程操作 DOM思路：我们在 Elements 里面找到输入框的 ID，使用 JQuery 操作。百度首页已经有 JQuery 了，其他网站我们可以在 Console 里执行 JS，加载一个。 我们切换到 Console 里直接用 JS 操作DOM，执行： $("input[name='wd']").val('测试'); $("form.fm").submit();相当于在百度输入框里输入了测试并点击了“百度一下”按钮。
现在再打开 http://127.0.0.1:9222/json ，可以看到原来的页面标题已经变成了 “测试_百度搜索”，也就说明成功完成了搜索。 获取搜索结果依然是在 Elements 里面找到结果列表的 ID，然后用 JS 获取内容。 获得结果数量： console.log($("#container").find(".nums").text());获得所有结果标题： $("#container").find(".c-container").each(function(){console.log($(this).find("h3.t").text())});
Chrome DevTools Protocol 的指令分为三十多个大类，每类又有若干个指令，这里不能一一介绍，只选择几个简单而常用的指令介绍一下： 跳转到指定页面：https://chromedevtools.github.io/devtools-protocol/tot/Page/执行JS函数：https://chromedevtools.github.io/devtools-protocol/tot/Runtime/获取资源树：https://chromedevtools.github.io/devtools-protocol/tot/Page/获取资源：https://chromedevtools.github.io/devtools-protocol/tot/Page/其中 Page.navigate 是必备指令，用于跳转页面。而 Runtime.evaluate 的效果等同于在 Develop Tools 的 Console 控制台执行指令，基本可以执行任何js指令，模拟输入，输出渲染后的 html 用它都可以轻松搞定，可以说是大杀器了。
获取资源树和获取资源指令则用于获取浏览器当前原始请求的数据，可以用它来构建Develop Tools的Source树。
可以说，利用 Develop Tools 实现的功能我们都可以通过 Chrome DevTools Protocol 实现，Chrome 自己也内置了一个官方的实现，用 Chrome 直接访问页面信息的 devtoolsFrontendUrl 即可看到，和按 F12 调用出来的 Develop Tools 基本一模一样。 3、chrome headless 爬虫抓取 websoket 数据From：https://brucedone.com/archives/1201 要分析的网站：https://datacenter.jin10.com/price Python 连接 websocket 代码： # coding:utf-8 from websocket import create_connection from websocket import ABNF api = "wss://sshibikfdn.jin10.com:9084/socket.io/?EIO=3&transport=websocket&sid=VsJvZikGdc8spBaPAAMO" headers = { 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', 'Cookie': 'UM_distinctid=16614315bf179b-0354a40c6714ff-34677908-232800-16614315bf2acb; XSRF-TOKEN=eyJpdiI6IkNZRU9uSmM1ZnY2M0VqNUttK1pxRGc9PSIsInZhbHVlIjoiRlJpNlRuekIxTDJZeVd3bHpvXC9OUEZGamw4VndZNEdXTEVsRjRMaFFyOEIxUHRtNDdTc1JaQ042eG4xdjlFeWJjWGlkcWFaeWl6NTRVUUlQMThaZmJ3PT0iLCJtYWMiOiJkYWU1MzQ2NjEyM2U3OTk0MzY5NWNjZTdhZmNlZjE0YTViMjc2YzBiYWM4YjhiMjNhZmRjMzU3YzliNDg3ZGIzIn0%3D; laravel_session=eyJpdiI6IjBRS3h0Y29XcGRBRlFIc0xIeWFiZGc9PSIsInZhbHVlIjoibVRLblpNTDJJa1JIN1ZJc0s5c2xrSkYzckNadDB6aGp0REd5SVJQTlkxNVAzajhvdXY5ZElSQ3VTcGVicjNiSXZ3NE9pZDZOdHJUM1d6WG1KQjZXNkE9PSIsIm1hYyI6Ijg3MWVkZDVlMDFjZDM2NDRjZmI2ZDhkNDJmZGI5MjNhMzk3MTViNmI1YTNmMDRmYWJjNzQ4ZGU2YWZhNzNhNzUifQ%3D%3D; io=VsJvZikGdc8spBaPAAMO', 'Host': 'sshibikfdn.jin10.com:9084', 'Connection': 'Upgrade', 'Origin': 'https://datacenter.jin10.com', 'Pragma': 'no-cache', 'Sec-WebSocket-Extensions': 'permessage-deflate; client_max_window_bits', 'Sec-WebSocket-Key': 'g4UA3smEJ0eGufMkyz7AOw==', 'Sec-WebSocket-Version': '13', 'Upgrade': 'websocket', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36', } def get_web_socket(): start_message = "2probe" ws = create_connection( api, header=headers, cookie=headers['Cookie'], #origin=headers['Origin'], #host=headers['Host'] ) frame = ABNF.create_frame("2probe", ABNF.OPCODE_TEXT) ws.send_frame(frame) data = ws.recv_frame() print(data) if __name__ == '__main__': get_web_socket() 使用 websocket 和 chrome header less 进行交互 import json import time import requests import websocket request_id = 0 target_url = 'https://datacenter.jin10.com/price' def get_websocket_connection(): r = requests.get('http://10.10.2.42:9222/json') #这是开启docker chrome headless的机器地址 if r.status_code != 200: raise ValueError("can not get the api ,please check if docker is ready") conn_api = r.json()[0].get('webSocketDebuggerUrl') return websocket.create_connection(conn_api) def run_command(conn, method, **kwargs): global request_id request_id += 1 command = {'method': method, 'id': request_id, 'params': kwargs} conn.send(json.dumps(command)) #while True: msg = json.loads(conn.recv()) if msg.get('id') == request_id: return msg def get_element(): conn = get_websocket_connection() msg = run_command(conn, 'Page.navigate', url=target_url) time.sleep(5) js = "var p = document.querySelector('.jin-pricewall_list-item_b').innerText ; p ;" for _ in range(20): time.sleep(1) msg = run_command(conn, 'Runtime.evaluate', expression=js) print(msg.get('result')['result']['value']) if __name__ == '__main__': get_element()整体逻辑非常简单，打开指定页面，等待页面数据刷新，然后直接偷懒拿数据渲染之后的页面值，运行效果如下：
这里使用 chrome-headless 的相关渲染环境来解决了抓取数据的问题，并且使用 websocket api 来进一步操作，其实 google 官方有 sdk 进行操作，https://github.com/GoogleChrome/puppeteer ，渲染的终究不是高效的做法，但是对于这种单页面目的性很强的数据，可以尝试渲染大法 Pyppeteer 防止检测 4、chrome devtools protocolchrome devtools protocol 允许第三方对基于 chrome 的 web 应用程序进行调试、分析等，它基于 WebSocket，利用 WebSocke t建立连接 DevTools 和浏览器内核的快速数据通道。一句话，有了这个协议就可以自己开发工具获取 chrome 的数据 协议详细内容看这里 chrome devtools protocol：Chrome DevTools Protocol 目前已经有很多大神针对这个协议封装出不同语言（nodejs，python，java...）的库。 awesome-chrome-devtools：https://github.com/ChromeDevTools/awesome-chrome-devtools Protocol Driver Libraries JavaScript/Node.js: chrome-remote-interface - The most-used JavaScript API for the protocolTypeScript/Node.js: chrome-debugging-clientJava: chrome-devtools-java-clientJava: karate - Web-service testing framework with a Java API to automate ChromeJava: jvppeteer - Headless Chrome For JavaPython: PyCDP - Pure-Python, sans-IO wrappers. See also the Trio CDP driverPython: chromewhip - drop-in replacement for the splash servicePython: pychrome - low level CDP transport handlerPython: ChromeController - high-level browser mgmtGo: chromedp - High-level actions and tasks for driving browsersGo: cdpGo: gcdGo: godetGo: RodC#/dotnet: chrome-dev-tools - Protocol wrapper generator that can be customized by editing handlebars templates. Includes .Net Core template.Ruby: Cuprite - Capybara driverRuby: ChromeRemoteKotlin: chrome-reactive-kotlin - reactive (rxjava 2.x), low-level client library in KotlinKotlin: chrome-devtools-kotlin - A coroutine-based client library, providing low-level CDP primitives and high-level extensions.Clojure: clj-chrome-devtools - The CDP wrapper API is autogenerated and will be updated when CDP protocol changes.PHP: chrome-devtools-protocol - A PHP client library for the protocol. PyChromeDevTools 爬虫 示例PyChromeDevTools：https://github.com/marty90/PyChromeDevTools PyChromeDevTools 是一个 Python 模块，允许在 Python 脚本中使用 Chrome DevTools 协议与 Google Chrome 交互。 页面加载时间 import PyChromeDevTools import time chrome = PyChromeDevTools.ChromeInterface() chrome.Network.enable() chrome.Page.enable() start_time=time.time() chrome.Page.navigate(url="http://www.google.com/") chrome.wait_event("Page.loadEventFired", timeout=60) end_time=time.time() print ("Page Loading Time:", end_time-start_time)打印所有安装的 cookies import PyChromeDevTools import time chrome = PyChromeDevTools.ChromeInterface() chrome.Network.enable() chrome.Page.enable() chrome.Page.navigate(url="http://www.nytimes.com/") chrome.wait_event("Page.frameStoppedLoading", timeout=60) #Wait last objects to load time.sleep(5) cookies,messages = chrome.Network.getCookies() for cookie in cookies["result"]["cookies"]: print ("Cookie:") print ("\tDomain:", cookie["domain"]) print ("\tKey:", cookie["name"]) print ("\tValue:", cookie["value"]) print ("\n")打印页面的所有对象 URL import PyChromeDevTools chrome = PyChromeDevTools.ChromeInterface() chrome.Network.enable() chrome.Page.enable() chrome.Page.navigate(url="http://www.facebook.com") event,messages=chrome.wait_event("Page.frameStoppedLoading", timeout=60) for m in messages: if "method" in m and m["method"] == "Network.responseReceived": try: url=m["params"]["response"]["url"] print (url) except: pass pychrome 爬虫 示例Pychrome：能跟 chrome 开发者工具交流的 Python 包，查看 github 代码中 examples 文件夹，查看更多例子。pychrome github地址，使用方法很简单，直接看 github上它的 Demo 这个库依赖 websocket-client python - pychrome 页面抓取测试：https://blog.csdn.net/max229max/article/details/91972429 使用 Chrome-headless 抓取页面内容，使用 python 的 pychrome 包。 要先开启浏览器，然后通过pychrome调用chrome dev protocol # -*- coding: utf-8 -*- # @Author : # @Date : 2021/8/19 # @File : test.py # @description : XXX import pychrome from concurrent.futures import ThreadPoolExecutor def request_will_be_sent(**kwargs): print("loading: %s" % kwargs.get('request').get('url')) def start_chrome(chrome_url='http://127.0.0.1:9222'): browser = pychrome.Browser(url=chrome_url) tab = browser.new_tab() tab.set_listener("Network.requestWillBeSent", request_will_be_sent) tab.start() tab.call_method("Network.enable") tab.call_method("Page.navigate", url="https://detail.tmall.com/item.htm?id=612274990981", _timeout=10) tab.wait(20) tab.stop() browser.close_tab(tab) if __name__ == '__main__': """ chrome.exe --remote-debugging-port=9222 --user-data-dir=c:/user_data_dir_9222 chrome.exe --remote-debugging-port=9223 --user-data-dir=c:/user_data_dir_9223 """ tem_list = ('http://127.0.0.1:9222', 'http://127.0.0.1:9223') with ThreadPoolExecutor() as tp_executor: tp_executor.submit(start_chrome, 'http://127.0.0.1:9222') tp_executor.submit(start_chrome, 'http://127.0.0.1:9223') # start_chrome() pass Web 性能自动化chrome devtools protocol --- Web 性能自动化实践介绍 ：https://testerhome.com/topics/15817 在测试 Web 页面加载时间时，可能会是这样的： 打开 chrome 浏览器。按 F12 打开开发者工具。在浏览器上打开要测试的页面查看开发者工具中 Network 面板的页面性能数据并记录或者 开发者 Console 面板运行 performance.timing 和 performance.getEntries() 收集数据performance 相关信息看这里 https://www.w3.org/TR/navigation-timing-2/#the-performancetiming-interface 几十上百个页面，每个版本都这样来，估计疯了，所以就想怎么把它做成自动化呢？ 获取 performance api 数据这里使用 Runtime Domain 中运行 JavaScript 脚本的 API Runtime.evaluate # 开始前先启动chrome，启动chrome必须带上参数`--remote-debugging-port=9222`开启远程调试否则无法与chrome交互 browser = pychrome.Browser('http://127.0.0.1:%d' % 9222) tab = browser.new_tab() tab.start() tab.Runtime.enable() tab.Page.navigate(url={你的页面地址}) # 设置等待页面加载完成的时间 tab.wait(10) # 运行js脚本 timing_remote_object = tab.Runtime.evaluate( expression='performance.timing' ) # 获取performance.timing结果数据 timing_properties = tab.Runtime.getProperties( objectId=timing_remote_object.get('result').get('objectId') ) timing = {} for item in timing_properties.get('result'): if item.get('value', {}).get('type') == 'number': timing[item.get('name')] = item.get('value').get('value') # 获取performance.getEntries()数据 entries_remote_object = tab.Runtime.evaluate( expression='performance.getEntries()' ) entries_properties = tab.Runtime.getProperties( objectId=entries_remote_object.get('result').get('objectId') ) entries_values = [] for item in entries_properties.get('result'): if item.get('name').isdigit(): url_timing_properties = tab.Runtime.getProperties( objectId=item.get('value').get('objectId') ) entries_value = {} for son_item in url_timing_properties.get('result'): if (son_item.get('value', {}).get('type') == 'number'or son_item.get('value', {}).get('type') == 'string'): entries_value[son_item.get('name')] = son_item.get('value').get('value') entries_values.append(entries_value) 获取 Network 数据实际上 performance.getEntries() 不会记录 404 的请求信息，另外当前页面通过 js 触发新 html 页面请求时它只会记录第一个页面的请求，在这些情况下就需要通过 Network Domain 的 API 来收集所有请求信息，先介绍用到的 API ： Network.requestWillBeSent 每个 http 请求发送前回调Network.responseReceived 首次接送到 http 响应时回调Network.loadingFinished 请求加载完成时回调Network.loadingFailed 请求加载失败时回调 # 封装上面4个事件对应的回调方法 class NetworkAPIImplemention(object): def __init__(self): self.request_dict = {} # 首个请求开始时间 self.start = None def request_will_be_sent(self, **kwargs): if self.start is None: self.start = time.time() dict_http = { 'url':kwargs.get('request').get('url'), 'start':kwargs.get('timestamp') } self.request_dict[kwargs.get('requestId')]=dict_http #print "loading:%s" % kwargs.get('request').get('url') def loading_finished(self, **kwargs): # 服务器返回code 例如404也是finished self.request_dict[kwargs.get('requestId')]['end'] = kwargs.get('timestamp') self.request_dict[kwargs.get('requestId')]['size'] = kwargs.get('encodedDataLength') def response_received(self, **kwargs): self.request_dict[kwargs.get('requestId')]['type'] = kwargs.get('type') self.request_dict[kwargs.get('requestId')]['response'] = kwargs.get('response') def loading_failed(self, **kwargs): self.request_dict[kwargs.get('requestId')]['end'] = kwargs.get('timestamp') self.request_dict[kwargs.get('requestId')]['error_text'] = kwargs.get('errorText') network_api = NetworkAPIImplemention() browser = pychrome.Browser('http://127.0.0.1:%d' % 9222) tab = browser.new_tab() # 绑定回调函数 tab.Network.requestWillBeSent = network_api.request_will_be_sent tab.Network.responseReceived = network_api.response_received tab.Network.loadingFinished = network_api.loading_finished tab.Network.loadingFailed = network_api.loading_failed tab.start() tab.Network.enable() tab.Runtime.enable() # 是否禁用缓存 if disable_cache: tab.Network.setCacheDisabled(cacheDisabled=True) tab.Page.navigate(url={你的页面地址}) tab.wait(10) tab.stop() self.browser.close_tab(tab) # 获取的所有url详细信息 print network_api.request_dict 监听 页面 事件有时候特别是一些复杂的页面，页面依赖 js 和 后端资源数据，并不是通常意义上页 loadEventEnd 事件触发完就表示页面加载完成，这种情况可能需要依赖开发打点。 这里以开发设计了一个 Loaded 事件为例： # 具体事件注册方式和注册时机询问开发，所谓注册时机即要求在js对象生成后注册，我们项目中page是在一个js文件中声明的，需要等这个js文件请求完成后再注册 # 这边使用Promise方式，这种方式awaitPromise参数必须是True js = """ new Promise((resolve, reject) => { page.getController().getPageEvent().addEventListener("Loaded", function(){ resolve(new Date().getTime()); }); }); """ custom_result = tab.Runtime.evaluate( expression=js, awaitPromise=True, timeout=timeout * 1000 ) print custom_result.get('result').get('value')有个坑 peformance.now()获取与 chrome 开发者工具协议一样类型的时间时，这个时间不准确，只好用 new Date().getTime() 写在最后 一开始是使用 nodejs 的 chrome-remote-interface，但是发现 Page.loadEventFired 回调后不会再记录请求，事实上有些页面仍然有请求没有完成，不懂是不是我使用姿势不对。。。 附赠 W3C 的一幅图
|
【本文地址】